package com.bungarus.offline;

import com.bungarus.busi.BusiProcessing;
import com.bungarus.busi.BusiProcessingException;
import com.bungarus.codec.protobuf.MessageProtobuf;
import com.bungarus.model.MessageBuilder;
import com.bungarus.model.RetryFailedMessage;
import org.apache.log4j.Logger;
import org.springframework.data.mongodb.core.BulkOperations;
import org.springframework.data.mongodb.core.FindAndModifyOptions;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.Update;
import org.springframework.util.StringUtils;

import java.util.*;
import java.util.stream.Collectors;

/**
 * Created by tang on 2018/12/26.
 */
public class OfflineMessagePersist extends BusiProcessing {
    Logger logger = Logger.getLogger(OfflineMessagePersist.class);

    private final static OfflineMessagePersist persist = new OfflineMessagePersist();
    private static MongoTemplate offlineStore;

    private OfflineMessagePersist() {
    }

    public static OfflineMessagePersist getInstance(MongoTemplate template) {
        OfflineMessagePersist.offlineStore = template;
        return persist;
    }

    public void doBulkBusiness(MessageProtobuf.Message offlineMsg, List<String> groupMembers, List<Object> onlineSessionList) {
        List<String> innerGM = new ArrayList<>();
        List<Object> innerSL = new ArrayList<>();
        innerGM.addAll(groupMembers);
        innerSL.addAll(onlineSessionList);

        BusiProcessing.busiTaskPool.execute(() -> {
            // get the offline members' ids
            Set<String> onlineMembers = new LinkedHashSet<>();
            for(Object onlineSession: innerSL) {
                Set userSession = (Set) onlineSession;
                for (Object s : userSession) {
                    String[] values = StringUtils.tokenizeToStringArray(s.toString(), ",");
                    String reciptId = values[0];
                    onlineMembers.add(reciptId);
                    // TODO: separate offline messages for multiple devices
                }
            }
            //TODO:retrive all member with multiple devices
            Set<String> offlineMembers = new LinkedHashSet<>();
            innerGM.forEach(offlineMembers::add);

            offlineMembers.removeAll(onlineMembers);

            //init two bulks for offline pointer and message
            BulkOperations ops = offlineStore.bulkOps(BulkOperations.BulkMode.UNORDERED, OfflinePointer.class.getSimpleName().toLowerCase());
            BulkOperations ops2 = offlineStore.bulkOps(BulkOperations.BulkMode.UNORDERED, OfflineMessage.class.getSimpleName().toLowerCase());

            offlineMembers.forEach(uid -> {

                Query query = new Query();
                query.addCriteria(Criteria.where("tenant_id").is(String.valueOf(offlineMsg.getHeader().getTenantID())));
                query.addCriteria(Criteria.where("uid").is(uid));
                query.addCriteria(Criteria.where("app").is(offlineMsg.getHeader().getApp()));

                OfflinePointer pointer = offlineStore.findOne(query, OfflinePointer.class, OfflinePointer.class.getSimpleName().toLowerCase());
                long pointerCounter = 0;
                if (null == pointer) {
                    // init the counter as 1 and pointer as 0
                    OfflinePointer pointerNew = new OfflinePointer(
                            String.valueOf(offlineMsg.getHeader().getTenantID()),
                            uid,
                            offlineMsg.getHeader().getApp(),
                            offlineMsg.getHeader().getDev(), //TODO: should be different for devices
                            1,
                            0);
                    ops.insert(pointerNew);
                    pointerCounter = 1;
                }
                else {
                    Update update = new Update();
                    update.inc("counter", 1); // counter autoincrement
                    ops.updateOne(query, update);
                    pointerCounter = pointer.getCounter() + 1;
                }

                List<MessageProtobuf.Extension> extensions = offlineMsg.getHeader().getExtensionsList();
                List<RetryFailedMessage.Expand> expands = extensions2Expands(extensions);

                // set certain user id about from and to for group message
                String fromUser = offlineMsg.getHeader().getFrom();
                String toUser = uid;

                RetryFailedMessage msg = RetryFailedMessage.newBuilder()
                        .id(Long.valueOf(offlineMsg.getHeader().getId()))
                        .type(offlineMsg.getHeader().getType())
                        .to(offlineMsg.getHeader().getTo())
                        .tenant(offlineMsg.getHeader().getTenantID())
                        .sentTime(offlineMsg.getHeader().getSentTime())
                        .app(offlineMsg.getHeader().getApp())
                        .content(offlineMsg.getBody().getContent())
                        .dev(offlineMsg.getHeader().getDev()) //TODO: should be different for devices
                        .extensions(expands)
                        .from(offlineMsg.getHeader().getFrom())
                        .gaction(offlineMsg.getHeader().getGaction())
                        .lang(offlineMsg.getHeader().getLang())
                        .priority(offlineMsg.getHeader().getPriority())
                        .built();

                OfflineMessage offlineMessage = new OfflineMessage();
                offlineMessage.setOfflineId(pointerCounter); // use pointer counter as autoincrement id of offline message
                offlineMessage.setFrom(fromUser);
                offlineMessage.setTo(toUser);
                offlineMessage.setMessage(msg);
                ops2.insert(offlineMessage);
            });
            ops.execute();
            ops2.execute();

            innerGM.clear();
            innerSL.clear();
        });
    }

    @Override
    public void doBusiness(Object o) throws BusiProcessingException {
        MessageProtobuf.Message offlineMsg = (MessageProtobuf.Message) o;
        BusiProcessing.busiTaskPool.execute(() -> {
            //1.To increase user's offline message counter by one
            OfflinePointer pointer;
            if(!offlineStore.collectionExists(OfflinePointer.class.getSimpleName().toLowerCase())) {
                offlineStore.createCollection(OfflinePointer.class.getSimpleName().toLowerCase());
            }

            Query query = new Query();
            query.addCriteria(Criteria.where("tenant_id").is(String.valueOf(offlineMsg.getHeader().getTenantID())));
            query.addCriteria(Criteria.where("uid").is(offlineMsg.getHeader().getTo()));
            query.addCriteria(Criteria.where("app").is(offlineMsg.getHeader().getApp()));

            boolean pointerExists = offlineStore.exists(query,
                        OfflinePointer.class,
                        OfflinePointer.class.getSimpleName().toLowerCase());

            if(!pointerExists) {
                // init the counter as 1 and pointer as 0
                pointer = new OfflinePointer(
                        String.valueOf(offlineMsg.getHeader().getTenantID()),
                        offlineMsg.getHeader().getTo(),
                        offlineMsg.getHeader().getApp(),
                        offlineMsg.getHeader().getDev(),1, 0);

                offlineStore.save(pointer, OfflinePointer.class.getSimpleName().toLowerCase());
            }
            else{
                Update update = new Update();
                update.inc("counter", 1); // counter autoincrement
                pointer = offlineStore.findAndModify(query, update,
                        FindAndModifyOptions.options().returnNew(true), //When true, returns the modified document
                        OfflinePointer.class,
                        OfflinePointer.class.getSimpleName().toLowerCase());
            }

            //2.To persist the offline message with offline counter
            List<MessageProtobuf.Extension> extensions = offlineMsg.getHeader().getExtensionsList();
            List<RetryFailedMessage.Expand> expands = extensions2Expands(extensions);

            // set certain user id about from and to for group message
            String fromUser = offlineMsg.getHeader().getFrom();
            String toUser = offlineMsg.getHeader().getTo();
            for(RetryFailedMessage.Expand expand:expands){
                 if(expand.getKey().equals(MessageBuilder.TAG_REAL_RECEIPT)){
                     toUser = expand.getValue();
                     break;
                }
            };

            RetryFailedMessage msg = RetryFailedMessage.newBuilder()
                    .id(Long.valueOf(offlineMsg.getHeader().getId()))
                    .type(offlineMsg.getHeader().getType())
                    .to(offlineMsg.getHeader().getTo())
                    .tenant(offlineMsg.getHeader().getTenantID())
                    .sentTime(offlineMsg.getHeader().getSentTime())
                    .app(offlineMsg.getHeader().getApp())
                    .content(offlineMsg.getBody().getContent())
                    .dev(offlineMsg.getHeader().getDev())
                    .extensions(expands)
                    .from(offlineMsg.getHeader().getFrom())
                    .gaction(offlineMsg.getHeader().getGaction())
                    .lang(offlineMsg.getHeader().getLang())
                    .priority(offlineMsg.getHeader().getPriority())
                    .built();

            OfflineMessage offlineMessage = new OfflineMessage();
            offlineMessage.setOfflineId(pointer.getCounter()); // use pointer counter as autoincrement id of offline message
            offlineMessage.setFrom(fromUser);
            offlineMessage.setTo(toUser);
            offlineMessage.setMessage(msg);

            offlineStore.save(offlineMessage, OfflineMessage.class.getSimpleName().toLowerCase());
        });
    }

    private List<RetryFailedMessage.Expand> extensions2Expands(List<MessageProtobuf.Extension> extensions) {
        List<RetryFailedMessage.Expand> expands = extensions.stream()
                .map(extension -> {
                    RetryFailedMessage.Expand expand = new RetryFailedMessage.Expand();
                    expand.setKey(extension.getKey());
                    expand.setValue(extension.getValue());
                    return expand;
                })
                .collect(Collectors.toList());
        return expands;
    }
}
